package pers.cz.javacvdemo.oss.service;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ThreadUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;
import pers.cz.javacvdemo.oss.model.EventRecord;
import pers.cz.javacvdemo.oss.model.QueueDataWrapper;
import pers.cz.javacvdemo.redis.RedisService;
import pers.cz.javacvdemo.utils.JsonUtils;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * @version 1.0
 * @description: 监听oss上传删除事件
 * @author: ChengZhi 0336
 * @create: 2025-08-14 09:46
 **/
@Component
@Slf4j
@RequiredArgsConstructor
public class OssEventListener implements InitializingBean {

	private final RedisService redisService;

	private final RingBufferWorkerFactory ringBufferWorkerFactory;

	private final static String EVENT_KEY = "minio-events";

	private final Set<String> videoExt = Set.of(".mp4", ".flv", ".avi", ".wmv", ".mkv");

	private volatile boolean listening = true;
	private void startListen() {

		Thread thread = new Thread(() -> {
			while (listening) {
				try {
					ThreadUtils.sleep(Duration.ofSeconds(60));
					Map<String, String> eventRecordMap = redisService.HASH.hGetAll(EVENT_KEY);
					log.info("minio-events size:{}", eventRecordMap.size());
					log.warn("minio-events:{}", eventRecordMap);
					List<QueueDataWrapper> list = eventRecordMap.entrySet().stream().map(entry -> {
						EventRecord eventRecord = JsonUtils.fromJson(entry.getValue(), EventRecord.class);
						return new QueueDataWrapper(entry.getKey(), eventRecord);
					}).toList();
					// 监听到之后如何保证不重复处理？并且处理失败的能够再次被处理？丢到本地队列后就将他移除，处理失败后，再插入到队列中
					list.forEach(queueDataWrapper -> {
						log.warn("监听到事件：{}", JsonUtils.toJson(queueDataWrapper));
						boolean isVideo = videoExt.stream().anyMatch(queueDataWrapper.getKey()::endsWith);
						if (isVideo) {
							BucketEventProducer messageProducer = ringBufferWorkerFactory.getMessageProducer();
							messageProducer.publish(queueDataWrapper.getKey(), queueDataWrapper.getEventRecord());
						}
						redisService.HASH.hDel(EVENT_KEY, queueDataWrapper.getKey());
					});
				} catch (Exception e) {
					log.error("监听失败！", e);
				}
			}
		});
		thread.setName("oss-event-listener");
		thread.setDaemon(true);
		thread.start();
	}

	public void stopListen() {
		this.listening = false;
	}
	@Override
	public void afterPropertiesSet() throws Exception {
		startListen();
		// 暂时只初始化一个消费者
		ringBufferWorkerFactory.initRingBuffer(new BucketEventHandler[] {new BucketEventHandler()});
		log.info("开始监听redis中的桶事件");
	}
}
